package com.zyx.hbasedemo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @author zyx
 * @since 2021/7/4 18:38
 * desc: HBase相关案例
 */
public class HbaseDemo {
    private Connection connection;
    private Admin admin;

    @Before
    public void init() throws Exception {
        // 1、获取客户端连接
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "linux201:2181,linux202:2181,linux203:2181");
        connection = ConnectionFactory.createConnection(conf);
        // 2、创建admin对象
        admin = connection.getAdmin();
    }

    /**
     * 创建命名空间
     */
    @Test
    public void createNameSpace() throws IOException {

        // 3、创建namespace
        NamespaceDescriptor namespace = NamespaceDescriptor.create("xxx").build();
        admin.createNamespace(namespace);

    }

    /**
     * 查看所有的namespace
     */
    @Test
    public void listNameSpace() throws Exception {

        // 3、查看
        NamespaceDescriptor[] list = admin.listNamespaceDescriptors();

        for (NamespaceDescriptor namespace : list) {
            System.out.println(namespace.getName());
        }
    }

    /**
     * 查看命名空间下所有表
     */
    @Test
    public void listNamespaceTables() throws Exception {

        List<TableDescriptor> tables = admin.listTableDescriptorsByNamespace("hbase".getBytes());

        for (TableDescriptor table : tables) {
            System.out.println(new String(table.getTableName().getName()));
        }
    }

    /**
     * 删除namespace
     */
    @Test
    public void dropNameSpace() throws Exception {

        // 1、获取命名空间下所有表
        List<TableDescriptor> tables = admin.listTableDescriptorsByNamespace("xxx".getBytes());
        // 2、删除命名空间下所有表
        for (TableDescriptor table : tables) {
            //禁用表
            admin.disableTable(table.getTableName());
            //删除表
            admin.deleteTable(table.getTableName());
        }
        // 3、删除命名空间
        admin.deleteNamespace("xxx");
    }

    /**
     * 创建表
     */
    @Test
    public void createTable() throws Exception {

        // 1、描述列簇
        ColumnFamilyDescriptor baseInfo = ColumnFamilyDescriptorBuilder.newBuilder("base_info".getBytes()).build();
        ColumnFamilyDescriptor extraInfo = ColumnFamilyDescriptorBuilder.newBuilder("extra_info".getBytes()).build();
        // 2、描述表
        TableName tableName = TableName.valueOf("bigdata:person1");
        TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
                .setColumnFamily(baseInfo)
                .setColumnFamily(extraInfo)
                .build();
        // 3、创建表
        byte[][] keys = {"11".getBytes(), "22".getBytes(), "33".getBytes(), "44".getBytes()};
        admin.createTable(tableDescriptor, keys);//.createTable(tableDescriptor);
    }

    /**
     * 修改表
     */
    @Test
    public void alterTable() throws Exception {
        // 1、描述列簇
        ColumnFamilyDescriptor baseInfo = ColumnFamilyDescriptorBuilder.newBuilder("base_info".getBytes()).setMaxVersions(3).setMinVersions(3).build();
        ColumnFamilyDescriptor extraInfo = ColumnFamilyDescriptorBuilder.newBuilder("extra_info".getBytes()).build();
        // 2、描述表
        TableName tableName = TableName.valueOf("bigdata:person");
        TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
                .setColumnFamily(baseInfo)
                .setColumnFamily(extraInfo)
                .build();
        admin.modifyTable(tableDescriptor);
    }

    /**
     * 删除表
     */
    @Test
    public void dropTable() throws Exception {
        // 判断表是否存在
        if (admin.tableExists(TableName.valueOf("user"))) {
            // 1、禁用表
            admin.disableTable(TableName.valueOf("user"));
            // 2、删除表
            admin.deleteTable(TableName.valueOf("user"));
        }
    }

    /**
     * 插入数据
     *
     * @throws Exception 连接异常
     */
    @Test
    public void put() throws Exception {

        // 1、获取Table对象
        Table table = connection.getTable(TableName.valueOf("bigdata:person"));
        // 2、封装数据
        // put 'bigdata:person','1001','base_info:name','zhangsan'
        Put put = new Put("1001".getBytes());
        put.addColumn("base_info".getBytes(), "name".getBytes(), "zhangsan".getBytes());
        put.addColumn("base_info".getBytes(), "age".getBytes(), Bytes.toBytes(20));
        // 3、插入
        table.put(put);
        // 4、关闭table对象
        table.close();
    }

    /**
     * 批量插入数据
     *
     * @throws Exception 连接异常
     */
    @Test
    public void putList() throws Exception {

        // 1、获取Table对象
        Table table = connection.getTable(TableName.valueOf("bigdata:person"));
        // 2、封装数据
        // put 'bigdata:person','1001','base_info:name','zhangsan'
        ArrayList<Put> putList = new ArrayList<Put>();
        for (int i = 1; i <= 10; i++) {
            Put put = new Put(("1001" + i).getBytes());
            put.addColumn("base_info".getBytes(), "name".getBytes(), ("zhangsan" + i).getBytes());
            put.addColumn("base_info".getBytes(), "age".getBytes(), Bytes.toBytes(20 + i));
            putList.add(put);
        }

        // 3、插入
        table.put(putList);
        // 4、关闭table对象
        table.close();
    }

    /**
     * 根据rowkey查询数据
     */
    @Test
    public void get() throws Exception {

        // 1、获取Table对象
        Table table = connection.getTable(TableName.valueOf("bigdata:person"));
        // 2、封装数据
        // get 'bigdata:person','1001'
        // 查询整行
        // Get get = new Get("1001".getBytes());
        // 查询某个列簇
        // Get get = new Get("1001".getBytes());
        // get.addFamily("base_info".getBytes());
        // 查询单个单元格的数据
        /*Get get = new Get("1001".getBytes());
        Result result = table.get(get);
        String name = new String(result.getValue("base_info".getBytes(), "name".getBytes()));*/
        // 查询某个列的数据
        Get get = new Get("1001".getBytes());
        get.addColumn("base_info".getBytes(), "name".getBytes());

        // 3、查询
        Result result = table.get(get);
        // 查询单个单元格的数据
        String name = new String(result.getValue("base_info".getBytes(), "name".getBytes()));
        List<Cell> cells = result.listCells();
        // 4、结果展示
        for (Cell cell : cells) {
            //[...,....,...,...,...]
            //列簇
            String family = new String(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
            //列限定符
            String qualifier = new String(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            //rowkey
            String rowkey = new String(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());

            if (family.equals("base_info") && qualifier.equals("age")) {
                int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("列簇=" + family + ",列限定符=" + qualifier + ",rowkey=" + rowkey + ",value=" + value);
            } else {

                String value = new String(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                System.out.println("列簇=" + family + ",列限定符=" + qualifier + ",rowkey=" + rowkey + ",value=" + value);
            }

        }
        // 5、关闭table
        table.close();
    }

    /**
     * 批量查询
     */
    @Test
    public void batchGet() throws Exception {

        // 1、获取Table对象
        Table table = connection.getTable(TableName.valueOf("bigdata:person"));
        // 2、封装数据
        // get 'bigdata:person','1001'
        // 查询整行
        // Get get = new Get("1001".getBytes());
        // 查询某个列簇
        // Get get = new Get("1001".getBytes());
        // get.addFamily("base_info".getBytes());
        // 查询某个列的数据
        ArrayList<Get> gets = new ArrayList<Get>();
        for (int i = 1; i <= 5; i++) {
            Get get = new Get(("1001" + i).getBytes());
            //get.addColumn("base_info".getBytes(),"name".getBytes());
            gets.add(get);
        }

        // 3、查询
        Result[] results = table.get(gets);
        for (Result result : results) {

            List<Cell> cells = result.listCells();
            // 4、结果展示
            for (Cell cell : cells) {
                // [...,....,...,...,...]
                // 列簇
                String family = new String(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
                // 列限定符
                String qualifier = new String(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                // rowkey
                String rowkey = new String(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());

                if (family.equals("base_info") && qualifier.equals("age")) {
                    int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列簇=" + family + ",列限定符=" + qualifier + ",rowkey=" + rowkey + ",value=" + value);
                } else {

                    String value = new String(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列簇=" + family + ",列限定符=" + qualifier + ",rowkey=" + rowkey + ",value=" + value);
                }


            }
        }

        // 5、关闭table
        table.close();
    }

    /**
     * 扫描
     */
    @Test
    public void scan() throws Exception {

        // 1、获取table对象
        Table table = connection.getTable(TableName.valueOf("bigdata:person"));
        // 2、获取scanner
        // scan 'bigdata:person'
        // 查询全部数据
        // Scan scan = new Scan();
        // ResultScanner scanner = table.getScanner(scan);
        // 查询某个列簇的数据
        /* Scan scan = new Scan();
        scan.addFamily("base_info".getBytes());
        ResultScanner scanner = table.getScanner(scan);*/
        // 查询某个列的数据
        // Scan scan = new Scan();
        // scan.addColumn("base_info".getBytes(),"name".getBytes());
        // ResultScanner scanner = table.getScanner(scan);
        // 查询某段rowkey范围的数据
        Scan scan = new Scan();
        scan.withStartRow("10011".getBytes(), true);
        scan.withStopRow("10014".getBytes(), true);
        ResultScanner scanner = table.getScanner(scan);
        //scanner.

        // 3、查询数据
        Iterator<Result> it = scanner.iterator();
        // 4、结果展示
        while (it.hasNext()) {
            Result result = it.next();
            List<Cell> cells = result.listCells();
            //4、结果展示
            for (Cell cell : cells) {
                /*//[...,....,...,...,...]
                // 列簇
                String family = new String(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
                // 列限定符
                String qualifier = new String(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                // rowkey
                String rowkey = new String(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());

                if(family.equals("base_info") && qualifier.equals("age")){
                    int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列簇="+family+",列限定符="+qualifier+",rowkey="+rowkey+",value="+value);
                }else {

                    String value = new String(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列簇="+family+",列限定符="+qualifier+",rowkey="+rowkey+",value="+value);
                }*/

                //列簇
                byte[] familyByte = CellUtil.cloneFamily(cell);
                String family = new String(familyByte);

                //列限定符
                String qualifier = new String(CellUtil.cloneQualifier(cell));

                //rowky
                String rowkey = new String(CellUtil.cloneRow(cell));

                //value
                if (family.equals("base_info") && qualifier.equals("age")) {
                    int value = Bytes.toInt(CellUtil.cloneValue(cell));
                    System.out.println("列簇=" + family + ",列限定符=" + qualifier + ",rowkey=" + rowkey + ",value=" + value);
                } else {
                    String value = new String(CellUtil.cloneValue(cell));
                    System.out.println("列簇=" + family + ",列限定符=" + qualifier + ",rowkey=" + rowkey + ",value=" + value);
                }


            }
        }

        // 5、table关闭
        table.close();
    }

    /**
     * select * from person where name like '%san%'
     */
    @Test
    public void queryLike() throws Exception {

        // 1、获取table
        Table table = connection.getTable(TableName.valueOf("bigdata:person"));
        // 2、封装数据、封装过滤器
        // 单个Cell过滤器，根据Cell value值进行过滤，只会显示符合要求的Cell数据
        Scan scan = new Scan();
        // SubstringComparator like = new SubstringComparator("2");
        // ValueFilter valueFilter = new ValueFilter(CompareOperator.EQUAL, like);
        // scan.setFilter(valueFilter);
        // 3、查询
        // select * from person where name='zhangsan2'
        // SingleColumnValueFilter filter = new SingleColumnValueFilter("base_info".getBytes(),"name".getBytes(),CompareOperator.EQUAL,"zhangsan2".getBytes());
        // scan.setFilter(filter);
        // select * from person where name like '%2%'
        SubstringComparator like = new SubstringComparator("2");
        SingleColumnValueFilter filter = new SingleColumnValueFilter("base_info".getBytes(), "name".getBytes(), CompareOperator.EQUAL, like);
        scan.setFilter(filter);
        ResultScanner results = table.getScanner(scan);
        Iterator<Result> it = results.iterator();
        // 4、结果展示
        while (it.hasNext()) {
            Result result = it.next();
            List<Cell> cells = result.listCells();
            //4、结果展示
            for (Cell cell : cells) {
                /*//[...,....,...,...,...]
                // 列簇
                String family = new String(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
                // 列限定符
                String qualifier = new String(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                // rowkey
                String rowkey = new String(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());

                if(family.equals("base_info") && qualifier.equals("age")){
                    int value = Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列簇="+family+",列限定符="+qualifier+",rowkey="+rowkey+",value="+value);
                }else {

                    String value = new String(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                    System.out.println("列簇="+family+",列限定符="+qualifier+",rowkey="+rowkey+",value="+value);
                }*/

                // 列簇
                byte[] familyByte = CellUtil.cloneFamily(cell);
                String family = new String(familyByte);

                // 列限定符
                String qualifier = new String(CellUtil.cloneQualifier(cell));

                // rowky
                String rowkey = new String(CellUtil.cloneRow(cell));

                // value
                if (family.equals("base_info") && qualifier.equals("age")) {
                    int value = Bytes.toInt(CellUtil.cloneValue(cell));
                    System.out.println("列簇=" + family + ",列限定符=" + qualifier + ",rowkey=" + rowkey + ",value=" + value);
                } else {
                    String value = new String(CellUtil.cloneValue(cell));
                    System.out.println("列簇=" + family + ",列限定符=" + qualifier + ",rowkey=" + rowkey + ",value=" + value);
                }


            }
        }

        // 5、table关闭
        table.close();
    }

    /**
     * 删除数据
     */
    @Test
    public void delete() throws Exception {

        // 1、获取table对象
        Table table = connection.getTable(TableName.valueOf("bigdata:person"));
        // 2、封装数据
        // delete/deleteall 'person','1001'
        Delete delete = new Delete("10011".getBytes());
        delete.addColumn("base_info".getBytes(), "age".getBytes());
        // 3、执行删除
        table.delete(delete);
        // 4、关闭table
        table.close();
    }

    @Test
    public void truncate() throws Exception {
        // 禁用表
        admin.disableTable(TableName.valueOf("bigdata:person"));
        // 清空表
        admin.truncateTable(TableName.valueOf("bigdata:person"), false);
    }

    @After
    public void close() throws Exception {
        //4、关闭连接
        if (admin != null)
            admin.close();
        if (connection != null)
            connection.close();
    }
}
